package com.bungarus.sdk;

import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.sdk.exception.ConnectionExceedMaxException;
import com.bungarus.sdk.exception.ConnectionFailedException;
import com.bungarus.sdk.listener.ClientMessageListener;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.log4j.Logger;

import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by tang on 2019/1/26.
 */
abstract public class Client implements Callback<EndUser, Void>{
    Logger logger = Logger.getLogger(Client.class);

    private static final long CONNECTION_INTERVAL = 5;
    private static final int MAX_RETRY = 3;

    public static final String ACCOUNT_CMD_PARAM = "-u";
    public static final String SEC_CMD_PARAM = "-p";
    //receipt of P2P chat
    public static final String RECEIPT_CMD_PARAM = "-t";
    //receipt of group chat
    public static final String GROUP_CMD_PARAM= "-g";

    protected static final NioEventLoopGroup busiTaskPool = new NioEventLoopGroup();

    private Bootstrap bootstrap;
    private NioEventLoopGroup worker = new NioEventLoopGroup();
    private Channel channel = null;

    protected EndUser loggedInUser;

    private CountDownLatch latch = new CountDownLatch(0);

    private List<ClientMessageListener> messageListeners = new LinkedList<>();

    public Client() {
        init();
    }

    private void init() {
        bootstrap = new Bootstrap();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new ClientChannelInitializer(this));
    }

    @Override
    public Void call(EndUser user) {
        this.loggedInUser = user;
        return null;
    }

    abstract public MessageProtobuf.Message makeloginMessage();

    public void doBusiness(MessageProtobuf.Message message) {
        if(MessageType.ENDUSER_RECONNECT == message.getHeader().getType()){
            String server = message.getHeader().getTo();
            if(null != server && server.length() > 0) {
                String[] remoteServer = server.split(":");
                assert remoteServer.length == 2;
                try {
                    connect(new InetSocketAddress(remoteServer[0], Integer.parseInt(remoteServer[1])),
                            value -> {
                                MessageProtobuf.Message loginMsg = makeloginMessage();
                                send(loginMsg, null);
                                return null;
                            });
                }
                catch (ConnectionExceedMaxException e) {
                    e.printStackTrace();
                }
            }
        }
        else {
            Client.busiTaskPool.execute(() -> {
                Client.this.messageListeners.stream().forEach(listener -> {
                    int type = message.getHeader().getType();
                    int retCode = message.getHeader().getReturnCode();
                    switch (type) {
                        case MessageType.STATUS:
                            listener.onStatus(message);
                            break;
                        case MessageType.IM:
                            listener.onIM(message);
                            break;
                        case MessageType.IM_GROUP:
                            listener.onGroupIM(message);
                            break;
                        case MessageType.ERROR: {
                            if(MessageStatus.ERR_AUTH_EXPIRATION == retCode) {
                                listener.onAuthExpire();
                            }
                            else if(MessageStatus.ERR_USER_NONEXIST == retCode) {
                                listener.onUserNotExist();
                            }
                            break;
                        }
                        default:
                            listener.onOthers(message);
                            break;
                    }
                });
            });
        }
    }

    public void connect(final InetSocketAddress server) throws ConnectionExceedMaxException {
        connect(server, 1, null);
    }

    public void connect(final InetSocketAddress server, final Callback<Void, Void> callback) throws ConnectionExceedMaxException {
        connect(server, 1, callback);
    }

    public void connect(final InetSocketAddress server, int retry, final Callback<Void, Void> callback) throws ConnectionExceedMaxException {
        if(null == server) {
            throw new IllegalArgumentException("The hostname or ip required.");
        }
        if(retry <= 0) {
            throw new IllegalArgumentException("Retry must be greater than 0.");
        }
        if(retry > MAX_RETRY) {
            String info = "Cannot connect to the server => " + server.toString() + ", exceeds the max retry number.";
            latch.countDown();
            throw new ConnectionExceedMaxException(info);
        }

        ChannelFuture future = bootstrap.connect(server);
        final int r = retry + 1;
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception{
                if (channelFuture.isSuccess()) {
                    Client.this.channel = channelFuture.sync().channel();
                    if(null != callback) {
                        callback.call(null);
                    }
                    latch.countDown();
                    logger.debug("Client connected to the server => " + server.toString());
                }
                else { //reconnect per 5 seconds
                    logger.debug("Client tries to reconnect to the server => " + server.toString());
                    Client.this.channel = null;
                    channelFuture.channel().eventLoop().schedule(
                            () -> {
                                try {
                                    connect(server, r, callback);
                                }
                                catch (ConnectionExceedMaxException e) {
                                    e.printStackTrace();
                                }
                            },
                            CONNECTION_INTERVAL,
                            TimeUnit.SECONDS);
                }
            }
        });
    }

    public void connect(List<InetSocketAddress> serverList) throws ConnectionFailedException {
        if(null != serverList && !serverList.isEmpty()) {
            for(int i=serverList.size()-1; i>=0; i--) {
                InetSocketAddress server = serverList.get(i);
                latch = new CountDownLatch(1);
                try {
                    connect(server, 1, null);
                    latch.await();
                    if(null != this.channel) {
                        break;
                    }
                    else {
                        serverList.remove(i);
                    }
                }
                catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(null == this.channel) {
                throw new ConnectionFailedException("All servers cannot be connected.");
            }
        }
    }

    public void send(MessageProtobuf.Message msg, Callback<Boolean, Void> result) {
        channel().writeAndFlush(msg).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if (channelFuture.isSuccess()) {
                    if(null != result) {
                        result.call(true);
                    }
                }
            }
        });
    }

    public void addListener(ClientMessageListener listener) {
        this.messageListeners.add(listener);
    }

    public void removeListener(ClientMessageListener listener) {
        this.messageListeners.remove(listener);
    }

    public List<ClientMessageListener> getListenerList() {
        return this.messageListeners;
    }

    public void close() {
        channel.close();
        messageListeners.clear();
        worker.shutdownGracefully();
    }

    public Channel channel() {
        return this.channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }
}
